package com.atguigu.tingshu.common.service;


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;

@Service
public class KafkaService {

    private static final Logger logger = LoggerFactory.getLogger(KafkaService.class);

    @Autowired
    private KafkaTemplate kafkaTemplate;

    public void sendMsg(String topic,String msg){
        this.sendMsg(topic,null,null,msg);
    }
    public void sendMsg(String topic,Integer partition,String key,String msg){
        CompletableFuture<SendResult> future = kafkaTemplate.send(topic, partition, key, msg);
        future.whenCompleteAsync((record,ex)->{
            if (ex!=null){
                ex.printStackTrace();
                throw new RuntimeException("发送消息到kafka失败!");
            }
        });
    }
}
